
use ssh2::Session;
use std::fs;
use std::io;
use std::io::Error;
use std::io::ErrorKind;
use std::io::{Read, Write};
use std::net::TcpStream;
use std::path::Path;
// use rand::Rand;
// use std::sync::mpsc::channel;
// use std::sync::mpsc::{Receiver, Sender};
use std::collections::HashMap;
use colored::Colorize;
use indicatif::MultiProgress;
use indicatif::ProgressBar;
use indicatif::ProgressStyle;
use log;
use rusqlite::{params, Connection};
use uuid::Uuid;
use std::error::Error as OError;

// use tokio::prelude::*;
// use tokio_core::net::TcpStream as ATcpStream;
// use async_ssh::Session as ASession;
// use futures::Future;

// use rusqlite::{params, Connection, Result};

// use time::Timespec;
const EA_TMP:&str = r#"
PY=python
if [ $(which python 1>/dev/null 2>&1 ; echo $? ) -ne 0 ];then
    PY=python3;
fi
function test_cmd {
    which $1 2>&1 1>/dev/null  && echo $?;
}
if [[ $(test_cmd ea) -eq 0 ]]; then
    ea -d stop;
fi
if [[ $(test_cmd ssserver ) -eq 0 ]]; then
    ssserver -d stop;
fi
RUN="yum"
if [[ $(test_cmd apt-get ) -eq 0 ]]; then
    RUN="apt-get";
fi
if [[ $(test_cmd git) -ne 0 ]]; then
    ps aux | grep apt-get | awk '{print $2 }' | xargs kill -9;
    $RUN install -y git;
fi
$RUN install -y python3 python3-pip libsodium-dev;
if [ -d /tmp/abc ];then
    rm -rf /tmp/abc
fi
cd /tmp/ && git clone https://github.com/f0cklinux/enuma-elish.git abc && cd abc ;
$PY -m enuma_elish.server -d stop;
$PY -m enuma_elish.server -c /tmp/ss.json -d start  ;rm /tmp/ss.json
echo $?"#;
type OutErr<T> = Result<T, Box<dyn OError>>;
// use threadpool;

pub trait AccountAction {
    fn bash(&mut self, cmd:&str);
    fn enuma_elish(&mut self);
    fn exec<HH>(&mut self, cmd: &str, f: HH)
    where
        HH: Fn(RemoteAccount) + 'static + Send + Copy;
    fn upload(&mut self, src: &str, dst: &str);
    fn download(&mut self, src: &str);
    fn down(&mut self, url: &str, out: Option<&str>);
    fn with<HH>(&mut self, f: HH)
    where
        HH: Fn(RemoteAccount) + 'static + Send + Copy;
    fn upload_exec<HH>(&mut self, src: &str, cmd: &str, f: HH)
    where
        HH: Fn(RemoteAccount) + 'static + Send + Copy;
    fn schedule_task(&self,oper :&str,time_class :&str,path :&str);
}
// Connect to the local SSH serer
pub struct AccountManager {
    db: Connection,
}

pub struct RemoteAccount {
    pub user: String,
    pub pwd: String,
    pub port: i32,
    use_key: bool,
    pub host: String,
    sess: Option<Session>,
    pub res: String,
    pub tag: String,
    pub loc: String,
}

impl Clone for RemoteAccount {
    fn clone(&self) -> Self {
        Self {
            user: self.user.clone(),
            pwd: self.pwd.clone(),
            port: self.port,
            use_key: self.use_key,
            host: self.host.clone(),
            sess: None,
            tag: self.tag.clone(),
            loc: self.loc.clone(),
            res: self.res.clone(),
        }
    }
}

impl Default for RemoteAccount {
    fn default() -> Self {
        Self {
            host: "127.0.0.1".to_string(),
            user: "root".to_string(),
            pwd: "".to_string(),
            port: 22,
            use_key: true,
            sess: None,
            res: "".to_string(),
            tag: String::new(),
            loc: String::new(),
        }
    }
}

impl RemoteAccount {
    pub fn addr(&self) -> String {
        format!("{}:{}", self.host, self.port)
    }
    #[allow(unused)]
    pub fn version(&mut self) -> String {
        self.exec("uname -a ")
    }

    pub fn create_ss_json(&mut self) -> String{
        let e_config = format!("/tmp/ss-random/{}", &self.host);
        if !Path::new(&e_config).parent().unwrap().exists(){
            let _ = fs::create_dir(Path::new(&e_config).parent().unwrap());
        }
        let pwd_base = Uuid::new_v4().to_string();
        let mut h:HashMap<i32, String> = HashMap::new();
        for i in 43440..43450{
            h.insert(i, format!("{}{}",pwd_base,i));
        }
        let ports_pwd = serde_json::json!(h);
        let s = serde_json::json!({
            "server": &self.host,
            "server_port":43443,
            "password":&format!("{}43443",pwd_base),
            "port_password": ports_pwd,
            "method":"aes-256-gcm"
        });
        if let Ok(mut f) = fs::File::create(&e_config){
            let _ = f.write_all(s.to_string().as_bytes());
        }
        e_config
    }

    pub fn check(&mut self, runners: Vec<&str>) {
        let s = self.exec("ls /bin  /usr/sbin /usr/bin /usr/local/bin /usr/sbin");
        let _ = runners
            .iter()
            .map(|x| {
                if s.contains(x) {
                    log::info!("{}|{} [{}]", self.host.yellow(), x, "O".green());
                } else {
                    log::info!("{}|{} [{}]", self.host.yellow(), x, "X".red());
                }
            })
            .collect::<Vec<_>>();
    }
    pub fn kill(&mut self, runners: &str) {
        self.exec(
            &format!(
            "ps aux | grep \"{}\" | grep -v '(grep|egrep)' | awk '++print $2--' | xargs kill -9 ",
            runners
        )
            .replace("++", "{")
            .replace("--", "}"),
        );
        if runners.starts_with(".") {
            log::error!(
                "danger to delted :{} so {}",
                runners.yellow(),
                "drop".on_red()
            );
        } else {
            self.exec(
                &format!("rm /tmp/{}.pid", runners.replace(" ", ""))
                    .replace("++", "{")
                    .replace("--", "}"),
            );
        }
    }

    pub fn exec(&mut self, cmd: &str) -> String {
        if let Some(sess) = &self.sess {
            let mut channel = sess.channel_session().unwrap();
            channel.exec(cmd).unwrap();
            let mut s = String::new();
            channel.read_to_string(&mut s).unwrap();
            let _ = channel.wait_close();
            self.res = s.clone();
            s
        } else {
            log::error!("{}", "no session".red());
            "".to_string()
        }
    }

    pub fn connect(&mut self) -> OutErr<()> {
        let tcp = TcpStream::connect(self.addr()).unwrap();
        let mut sess = Session::new()?;
        sess.set_tcp_stream(tcp);
        sess.handshake()?;
        let res = if self.pwd.len() > 0 {
            sess.userauth_password(&self.user, &self.pwd)
        } else {
            sess.userauth_agent(&self.user)
        };
        match res {
            Ok(_) => {
                sess.set_compress(true);
                self.sess = Some(sess);
                Ok(())
            }
            Err(e) => {
                log::error!("{}", format!("{}", e).red());
                Err(Box::new(Error::new(ErrorKind::Other, format!("{}", e))))
            }
        }
    }

    pub fn down(&self, url: &str, out: Option<&str>) {
        if let Some(sess) = &self.sess {
            let mut channel = sess.channel_session().unwrap();
            if let Some(out) = out {
                channel
                    .exec(&format!(
                        "wget -c -t 10  \"{url}\" -O {out} || curl -s \"{url}\" -o {out}",
                        url = url,
                        out = out
                    ))
                    .unwrap();
            } else {
                channel
                    .exec(&format!(
                        "wget -c -t 10  \"{url}\" -O {out} || curl -s \"{url}\" -o {out}",
                        url = url,
                        out = "/tmp/out"
                    ))
                    .unwrap();
            }

            let _ = channel.wait_close();
        }
    }

    pub fn download_file(
        &self,
        path: &Path,
        dst: Option<&str>,
        pro: Option<ProgressBar>,
    ) -> String {
        if let Some(sess) = &self.sess {
            let local_file_path = if let Some(s) = dst {
                s.to_string()
            } else {
                let id = Uuid::new_v4().to_string();
                format!("/tmp/{}", &id)
            };
            let mut local_writer = fs::File::create(&local_file_path).unwrap();
            let (mut remote_file, stat) = sess.scp_recv(path).unwrap();

            let pb = if let Some(progress) = pro {
                // progress.len = stat.size();
                progress
            } else {
                let ppb = ProgressBar::new(stat.size());
                ppb.set_style(ProgressStyle::default_bar()
                    .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {bytes_per_sec} ({eta})")
                    .progress_chars("#>-"));
                ppb
            };

            match copy_with_progress(&mut remote_file, &mut local_writer, &pb) {
                Ok(_) => {}
                Err(e) => {
                    log::error!("upload failed reason: {}", e);
                }
            }
            local_file_path.to_string()
        } else {
            log::error!("{}", "no session".red());
            "".to_string()
        }
    }

    pub fn upload_file(&self, path: &Path, dst: Option<&str>, pro: Option<ProgressBar>) -> String {
        if !path.exists() {
            log::error!("{} : [{}]", path.to_str().unwrap(), "failed".red());
            return "".to_string();
        }
        if let Some(sess) = &self.sess {
            let meta = fs::metadata(path).unwrap();
            let pb = if let Some(progress) = pro {
                progress
            } else {
                let ppb = ProgressBar::new(meta.len());
                ppb.set_style(ProgressStyle::default_bar()
                    .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {bytes_per_sec} ({eta})")
                    .progress_chars("#>-"));
                ppb
            };
            let remote_file_path = if let Some(s) = dst {
                s.to_string()
            } else {
                let id = Uuid::new_v4().to_string();
                format!("/tmp/{}", &id)
            };
            // pb.println(&format!("read {}",meta.len() ));
            let mut reader = fs::File::open(path).unwrap();
            let remote_path = Path::new(&remote_file_path);
            let mut remote_file = sess.scp_send(remote_path, 0o666, meta.len(), None).unwrap();
            match copy_with_progress(&mut reader, &mut remote_file, &pb) {
                Ok(size) => {
                    pb.finish_with_message(format!(
                        "{} upload:{} |{}| [{}]",
                        self.addr().yellow(),
                        path.to_str().unwrap(),
                        size,
                        "suss".green()
                    ));
                    // std::thread::sleep(std::time::Duration::from_secs(4));
                }
                Err(e) => {
                    pb.finish_with_message(format!(
                        "{} upload:{} |{}| [{}]",
                        self.addr().yellow(),
                        path.to_str().unwrap(),
                        "unknow",
                        "failed".green()
                    ));
                    log::error!("upload failed reason: {}", e);
                }
            };

            remote_file_path
        } else {
            log::error!("{}", "no session".red());
            "".to_string()
        }
    }
    #[allow(unused)]
    pub fn copy_self(&mut self, path: &Path, pro: Option<ProgressBar>) -> String {
        if path.exists() {
            let exe_path = self.upload_file(path, None, pro);
            if exe_path != "no session" {
                let pid = self.exec(&format!("{} --daemon -s \"0.0.0.0:54321\"", &exe_path));
                if pid != "no session" {
                    return pid;
                }
            }
        }
        return "copy self failed".to_string();
    }
}

impl AccountManager {
    pub fn new(db_path: &Path) -> Option<Self> {
        if db_path.exists() {
            Some(Self {
                db: Connection::open(db_path).unwrap(),
            })
        } else {
            None
        }
    }
    pub fn search_host(&self, host_tag: &str) -> Vec<RemoteAccount> {
        let mut stmt = self
            .db
            .prepare(&format!(
                "SELECT host, user, passwd, tag, location FROM Host where host like \"%{}%\" or tag like \"%{}%\" or location like \"%{}%\" ;",
                host_tag,
                host_tag,
                host_tag,
            ))
            .unwrap();
        let person_iter = stmt
            .query_map(params![], |row| {
                // let s: String = row.get(0).unwrap();
                // log::info!("{}", &s);
                Ok(RemoteAccount {
                    host: row.get(0).expect("get host failed"),
                    user: row.get(1).expect("get user failed"),
                    pwd: row.get(2).expect("get pwd failed"),
                    tag: row.get(3).expect("get tag failed"),
                    loc: row.get(4).expect("get loc failed"),
                    ..RemoteAccount::default()
                })
            })
            .expect("query map error");
        let mut v: Vec<RemoteAccount> = Vec::new();
        for p in person_iter {
            v.push(p.expect("iter Account error"));
        }
        v
    }
}

impl AccountAction for Vec<RemoteAccount>
{
    fn bash(&mut self, cmd:&str){
        let pool = threadpool::ThreadPool::new(30);
        for acc in self {
            let mut acc_shared = acc.clone();
            let cmd_str_shared = cmd.to_string();
            pool.execute(move || {
                match acc_shared.connect() {
                    Ok(_) => {
                        let res = acc_shared.exec(&cmd_str_shared);
                        log::info!("{}|{}",acc_shared.host.green(), res);
                    }
                    Err(e) => {
                        log::info!("{}|{}", &acc_shared.addr().red(), e);
                    }
                }
            });
        }
        pool.join();
    }

    
    fn schedule_task(&self,oper :&str, time_class: &str, schedule_script: &str){
        let pool = threadpool::ThreadPool::new(16);
        // self.upload_exec()
        // let dst_scripts = Path::join(Path::new(".every-scripts"),Path::new(schedule_script.split("/").last().unwrap())).to_str().unwrap().to_string();
        let m = MultiProgress::new();
        
        let sty = ProgressStyle::default_bar()
            .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {bytes_per_sec} ({eta}) {msg}")
            .progress_chars("#>-");

        for acc in self {
            let mut acc_shared = acc.clone();
            let src = schedule_script.clone();
            // let dst_scripts_share = dst_scripts.clone();
            let dst_scripts_share = Path::join(Path::new(".every-scripts"),Path::new(src.split("/").last().unwrap())).to_str().unwrap().to_string();
         
            let time_str_shared = time_class.to_string();
            let script_str_shared = schedule_script.to_string();
            
            let size = if Path::new(&script_str_shared).exists() {
                let ms = fs::metadata(&script_str_shared).unwrap();
                let size = ms.len();
                size
            } else {
                0
            };
            let oper_shared = oper.to_string();
            let pro_bar = m.add(ProgressBar::new(size));
            pro_bar.set_style(sty.clone());
            
            
            pool.execute(move || {
                match acc_shared.connect() {
                    Ok(_) => {
                        match oper_shared.as_ref(){
                            "add" => {
                                acc_shared.exec("mkdir -p ~/.every-scripts");
                                acc_shared.upload_file(&Path::new(&script_str_shared), Some(dst_scripts_share.as_ref()), Some(pro_bar));
                                match time_str_shared.as_ref() {
                                    "minute" => acc_shared.exec(&format!(r#"crontab -l 2>/dev/null ; echo "* * * * * /bin/sh $HOME/{}" | crontab -  "#, &dst_scripts_share)),
                                    "hour" =>{
                                        acc_shared.exec(&format!(r#"crontab -l 2>/dev/null ; echo "0 * * * * /bin/sh $HOME/{}" | crontab -  "#, &dst_scripts_share))
                                    },
                                    "day" => {
                                            acc_shared.exec(&format!(r#"crontab -l 2>/dev/null ; echo "0 23 * * * /bin/sh $HOME/{}" | crontab -  "#, &dst_scripts_share))
                                        },
                                    _ => {
                                        "".to_string()
                                    },
                                };
                                log::info!("add {}| {}",acc_shared.host.green(), &dst_scripts_share);
                            
                            },
                            "del" => {
                                acc_shared.exec(&format!(r#"crontab -l | grep -v "{}" | crontab - ; "#, &dst_scripts_share));
                                acc_shared.exec(&format!(r#"rm $HOME/{} ;"#, &dst_scripts_share));
                                log::info!("{} | {}", acc_shared.host.green(), &dst_scripts_share);
                            },
                            "start" => {
                                acc_shared.exec(&format!(r#"service cron start"#));
                                
                                log::info!("Start {} | {}", acc_shared.host.green(), &dst_scripts_share);
                            },
                            "stop" => {
                                acc_shared.exec(&format!(r#"service cron stop"#));
                    
                                log::info!("stop {} | {}", acc_shared.host.green(), &dst_scripts_share);
                            },
                            _ => {
                                log::info!(" Err {} | {}", acc_shared.host.green(), &dst_scripts_share);
                            }
                        };
                    }
                    Err(e) => {
                        log::info!("{}|{}", &acc_shared.addr().red(), e);
                    }
                }
            });
        }
        pool.join();
        
    }

    fn enuma_elish(&mut self){
        let pool = threadpool::ThreadPool::new(20);
        let m = MultiProgress::new();
        let sty = ProgressStyle::default_bar()
            .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {bytes_per_sec} ({eta}) {msg}")
            .progress_chars("#>-");
        
        for acc in self{
            let mut acc_shared = acc.clone();
            let e_config = acc.create_ss_json();
            let size = fs::metadata(&e_config).unwrap().len();
            let pro_bar = m.add(ProgressBar::new(size));
                pro_bar.set_style(sty.clone());
                
            pool.execute(move || {
                match acc_shared.connect() {
                    Ok(_) => {
                        acc_shared.upload_file(&Path::new(&e_config), Some("/tmp/ss.json"), Some(pro_bar));
                        std::thread::sleep(std::time::Duration::from_secs(3));
                        println!("{} : building",acc_shared.host.yellow());
                        let res = acc_shared.exec(&EA_TMP);
                        // (&m).println("");
                        if res.ends_with("0") || res.ends_with("0\n"){
                            println!("{} finish",&acc_shared.host.green());
                        }
                    },
                    Err(e) => {
                        log::info!("{}|{}", &acc_shared.addr().red(), e);
                    }
                }
                
            });

            
            m.join_and_clear().unwrap();
            pool.join();
        }
    }

    fn down(&mut self, url: &str, out: Option<&str>) {
        let pool = threadpool::ThreadPool::new(20);
        let u = url.to_string();
        let s = match out {
            Some(a) => a.to_string(),
            _ => "".to_string(),
        };
        for acc in self {
            let uu = u.clone();
            let oo = s.clone();

            let acc_shared = acc.clone();
            pool.execute(move || {
                let m = if oo.len() == 0 {
                    None
                } else {
                    Some(oo.as_str())
                };
                acc_shared.down(&uu, m);
            });
        }
    }
    fn with<HH>(&mut self, f: HH)
    where
        HH: Fn(RemoteAccount) + 'static + Send + Copy,
    {
        let pool = threadpool::ThreadPool::new(10);

        for acc in self {
            // let ff = f.clone();
            let mut acc_shared = acc.clone();
            // let c:&'a str = cmd.to_string();
            pool.execute(move || {
                // let cc = c.clone();
                match acc_shared.connect() {
                    Ok(_) => {
                        f(acc_shared.clone());
                    }
                    Err(e) => {
                        log::info!("{}|{}", &acc_shared.addr().red(), e);
                    }
                }
            });
        }
        pool.join();
    }

    fn exec<HH>(&mut self, cmd: &str, f: HH)
    where
        HH: Fn(RemoteAccount) + 'static + Send + Copy,
    {
        let pool = threadpool::ThreadPool::new(10);
        let cmd_string = cmd.to_string();
        for acc in self {
            let mut acc_shared = acc.clone();
            // let c:&'a str = cmd.to_string();
            let cmd_str_shared = cmd_string.clone();
            pool.execute(move || {
                // let cc = c.clone();
                match acc_shared.connect() {
                    Ok(_) => {
                        acc_shared.exec(&cmd_str_shared);
                        f(acc_shared.clone());
                    }
                    Err(e) => {
                        log::info!("{}|{}", &acc_shared.addr().red(), e);
                    }
                }
            });
        }
        pool.join();
    }

    fn upload_exec<HH>(&mut self, src: &str, cmd: &str, f: HH)
    where
        HH: Fn(RemoteAccount) + 'static + Send + Copy,
    {
        let pool = threadpool::ThreadPool::new(10);
        let m = MultiProgress::new();
        let sty = ProgressStyle::default_bar()
            .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {bytes_per_sec} ({eta}) {msg}")
            .progress_chars("#>-");

        let src = src.to_string();
        for acc in self {
            let mut acc_shared = acc.clone();
            let srcs = src.clone();
            let size = if Path::new(&srcs).exists() {
                let ms = fs::metadata(&srcs).unwrap();
                let size = ms.len();
                size
            } else {
                continue;
            };
            let pro_bar = m.add(ProgressBar::new(size));
            pro_bar.set_style(sty.clone());
            let cmd_copy = cmd.to_string().clone();
            pool.execute(move || {
                let p = Path::new(&srcs);

                match acc_shared.connect() {
                    Ok(_) => {
                        let exe_path = acc_shared.upload_file(p.as_ref(), None, Some(pro_bar));
                        acc_shared.exec(&format!("chmod +x {}", &exe_path));
                        acc_shared.exec(&format!("{} {}", &exe_path, &cmd_copy));
                        f(acc_shared);
                    }
                    Err(e) => {
                        log::info!("{}|{}", &acc_shared.addr().red(), e);
                    }
                };
            });
        }
        std::thread::sleep(std::time::Duration::from_secs(1));
        m.join_and_clear().unwrap();
        pool.join();
    }
    fn download(&mut self, src: &str) {
        let pool = threadpool::ThreadPool::new(10);
        // let m = MultiProgress::new();
        // let sty = ProgressStyle::default_bar()
        //     .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {bytes_per_sec} ({eta}) {msg}")
        //     .progress_chars("#>-");

        let src = src.to_string();
        for acc in self {
            let mut acc_shared = acc.clone();
            let srcs = src.clone();

            // let pro_bar = m.add(ProgressBar::new(100));
            // pro_bar.set_style(sty.clone());
            pool.execute(move || {
                let p = Path::new(&srcs);

                match acc_shared.connect() {
                    Ok(_) => {
                        let output = acc_shared.download_file(p.as_ref(), None, None);
                        log::info!("{} -> {}", p.to_str().unwrap().green(), output.green());
                    }
                    Err(e) => {
                        log::info!("{}|{}", &acc_shared.addr().red(), e);
                    }
                };
            });
        }
        // m.join_and_clear().unwrap();
        pool.join();
    }


    fn upload(&mut self, src: &str, dst: &str) {
        let pool = threadpool::ThreadPool::new(10);
        let m = MultiProgress::new();
        let sty = ProgressStyle::default_bar()
            .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {bytes_per_sec} ({eta}) {msg}")
            .progress_chars("#>-");

        let src = src.to_string();
        for acc in self {
            let mut acc_shared = acc.clone();
            let dst = dst.to_string();
            let srcs = src.clone();
            let size = if Path::new(&srcs).exists() {
                let ms = fs::metadata(&srcs).unwrap();
                let size = ms.len();
                size
            } else {
                continue;
            };
            let pro_bar = m.add(ProgressBar::new(size));
            pro_bar.set_style(sty.clone());
            pool.execute(move || {
                let p = Path::new(&srcs);

                match acc_shared.connect() {
                    Ok(_) => {
                        if dst.len() == 0 {
                            acc_shared.upload_file(p.as_ref(), None, Some(pro_bar));
                        } else {
                            let r_p = Path::new(&dst);
                            acc_shared.exec(&format!(
                                "mkdir -p {}",
                                r_p.parent().unwrap().to_str().unwrap()
                            ));
                            acc_shared.upload_file(p.as_ref(), Some(&dst), Some(pro_bar));
                        }
                        std::thread::sleep(std::time::Duration::from_secs(3));
                    }
                    Err(e) => {
                        log::info!("{}|{}", &acc_shared.addr().red(), e);
                    }
                };
            });
        }

        m.join_and_clear().unwrap();
        pool.join();
    }
}

fn copy_with_progress<R: ?Sized, W: ?Sized>(
    reader: &mut R,
    writer: &mut W,
    pro: &ProgressBar,
) -> io::Result<u64>
where
    R: Read,
    W: Write,
{
    let mut buf = [0; 54321];
    let mut written = 0;
    // let mut wr = 0;
    loop {
        // pro.println(&format!("read wait "));
        let len = match reader.read(&mut buf) {
            Ok(0) => return Ok(written),
            Ok(len) => len,
            // Err(ref e ) if e.kind() == io::ErrorKind::Interrupted => continue,
            Err(e) => {
                pro.println(&format!("{}", e));
                return Err(e);
            }
        };

        let mut ws = writer.write(&buf[..len]).expect("write error ");
        if ws < 54321 {
            'some: loop {
                ws += writer.write(&buf[ws..len]).expect("write error ");
                if ws == len {
                    break 'some;
                }
            }
        }

        written += len as u64;
        // wr += ws;
        // pro.println(&format!("{} {}\r",written, wr ));
        pro.inc(len as u64);
        buf = [0; 54321];
    }
}
